package demo;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

@Slf4j
public class MovieClassifyMapper extends Mapper<LongWritable, Text, Text, DistanceAndLabel> {
    private DistanceAndLabel distance_label = new DistanceAndLabel();
    private String splitter = "";
    ArrayList<String> testData = new ArrayList<String>();
    private String testPath = "";

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, DistanceAndLabel>.Context context)
            throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();
        splitter = conf.get("SPLITTER");
        testPath = conf.get("TESTPATH");
        //读取测试数据存于列表testData中
        FileSystem fs = FileSystem.get(conf);
        FSDataInputStream is = fs.open(new Path(testPath));
        BufferedReader br = new BufferedReader(new InputStreamReader(is));
        String line = "";
        while ((line = br.readLine()) != null) {
            testData.add(line);
        }
        is.close();
        br.close();
    }

    /**
     * map阶段，setup函数读取测试数据。在map函数里读取每条训练数据，遍历测试数据，
     * 计算读取进来的训练记录与每条测试数据的距离，计算距离采用的是欧式距离的计算方法，
     * map输出的键是每条测试数据，输出的值是该测试数据与读取的训练数据的距离和训练数据的类别。
     *
     * 读取训练数据，计算该训练数据与每条测试数据的距离，输出《测试数据，距离及训练数据的类别》
     *
     * @param key
     * @param value   训练数据集
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DistanceAndLabel>.Context context)
            throws IOException, InterruptedException {
        double distance = 0.0;
        String[] val = value.toString().split(splitter);
        //gender_genre 1,1,1,10,48067    ,0,5,5,0,6,2,0,0,3,21,3,20,2,3,18,14,0,14
        String[] singleTrainData = Arrays.copyOfRange(val, 5, val.length);
        String sex = val[1];//性别
        for (String td : testData) {
            String[] test = td.split(splitter);
            String[] singleTestData = Arrays.copyOfRange(test, 5, test.length);
            distance = Distance(singleTrainData, singleTestData);
            distance_label.setDistance(distance);
            distance_label.setLabel(sex);

            //输出《测试数据，距离及训练数据的类别》
            log.info("《测试数据，距离及训练数据的类别》:{} -> {}",td,distance_label);
            context.write(new Text(td), distance_label);
        }
    }

    /**
     * 计算训练数据与测试数据的距离
     *
     * @param singleTrainData
     * @param singleTestData
     * @return
     */
    private double Distance(String[] singleTrainData, String[] singleTestData) {
        double sum = 0.0;
        for (int i = 0; i < singleTrainData.length; i++) {
            sum += Math.pow(Double.parseDouble(singleTrainData[i]), Double.parseDouble(singleTestData[i]));
        }
        return Math.sqrt(sum);
    }
}
